#include "signaling_worker.h"

#include <rtc_base/logging.h>
#include <rtc_base/zmalloc.h>

#include "base/socket.h"
#include "base/xhead.h"
#include "rtc_server.h"

extern xrtc::RtcServer* g_rtc_server;

namespace xrtc {

// 处理管道写入的回调函数    
void signalingWorkerRecvNotify(EventLoop* /* el */, IOWatcher* /* w */, int fd, int /* events */, void *data) {
    int msg;
    if (read(fd, &msg, sizeof(int)) != sizeof(int)) {
        RTC_LOG(LS_WARNING) << "read from pipe error: " << strerror(errno) << ", errno: " << errno;
        return;
    }

    SignalingWorker *worker = (SignalingWorker*)(data); 
    worker->_processNotify(msg);
}

// 处理连接读/写事件的回调函数
void conn_io_cb(EventLoop *, IOWatcher *, int fd, int events, void *data) {
    SignalingWorker *worker =  (SignalingWorker*)(data); 
    if (events & EventLoop::READ) {
        worker->_readQuery(fd);
    }

    if (events & EventLoop::WRITE) {
        worker->_write_reply(fd);
    }
}


// 处理连接超时的回调函数
void conn_time_cb(EventLoop* el, TimerWatcher* /*w*/, void* data) {
    SignalingWorker *worker =  (SignalingWorker*)(el->owner()); 
    TcpConnection* conn = (TcpConnection*)data;
    
    worker->_processTimeout(conn);
}

// 创建事件循环
SignalingWorker::SignalingWorker(int worker_id, const SignalingServerOptions& options) 
                : m_worker_id(worker_id), m_options(options), m_el(new EventLoop(this)) {

}

SignalingWorker::~SignalingWorker() {
    for (auto conn : m_connections) {
        if (conn) {
            _closeConn(conn);
        }
    }

    m_connections.clear();

    if (m_el) {
        delete m_el;
        m_el = nullptr;
    }

    if (m_thread) {
        m_thread.reset();
    }
}

int SignalingWorker::init() {
    int fds[2];
    if (pipe(fds)) {
        RTC_LOG(LS_WARNING) << "create pipe error: " << strerror(errno) << ", errno: " << errno;
        return -1;
    }

    m_notify_recv_fd = fds[0];
    m_notify_send_fd = fds[1];

    // 注册管道可读事件的回调函数
    m_pipe_watcher = m_el->createIOEvent(signalingWorkerRecvNotify, this);
    m_el->startIOEvent(m_pipe_watcher, m_notify_recv_fd, EventLoop::READ);

    return 0;
}

// 创建工作线程
bool SignalingWorker::start() {
    if (m_thread) {
        RTC_LOG(LS_WARNING) << "signaling worker already start, worker_id: " << m_worker_id;
        return false;
    }

    m_thread = std::make_unique<std::thread>(([=]() {
        RTC_LOG(LS_INFO) << "signaling worker event loop start, worker_id: " << m_worker_id;
        m_el->start();
        RTC_LOG(LS_INFO) << "signaling worker event loop stop, worker_id: " << m_worker_id;
    }));

    return true;
}

// 发送信号通知线程退出
void SignalingWorker::stop() {
    notify(SignalingWorker::QUIT);
}

int SignalingWorker::notify(int msg) {
    int written = write(m_notify_send_fd, &msg, sizeof(int));
    return written == sizeof(int) ? 0 : -1;
}

void SignalingWorker::join() {
    if (m_thread && m_thread->joinable()) {
        m_thread->join();
    }
}


// 通知线程有新连接
int SignalingWorker::notifyNewConn(int fd) {
    m_lock_queue.produce(fd);
    return notify(SignalingWorker::NEW_CONN);
}

void SignalingWorker::pushMsg(std::shared_ptr<RtcMsg> msg) {
    std::lock_guard<std::mutex> lock(m_que_msg_mtx);
    m_que_msg.push(msg);
}

std::shared_ptr<RtcMsg> SignalingWorker::popMsg() {
    std::lock_guard<std::mutex> lock(m_que_msg_mtx);
    if (m_que_msg.empty()) {
        return nullptr;
    }

    std::shared_ptr<RtcMsg> msg = m_que_msg.front();
    m_que_msg.pop();

    return msg;
}

int SignalingWorker::sendRtcMsg(std::shared_ptr<RtcMsg> msg) {
    pushMsg(msg);
    return notify(RTC_MSG);
}

void SignalingWorker::_processNotify(int msg) {
    switch (msg) {
        case QUIT:
            _stop();
            break;

        case NEW_CONN:
            int fd;
            if (m_lock_queue.consume(&fd)) {
                _newConn(fd);
            }
            break;
                
        case RTC_MSG:
            _processRtcMsg();
            break;        

        default:
            RTC_LOG(LS_WARNING) << "unknown notify msg: " << msg;
            break;
    }
}


// 设置新连接的socket配置, 创建TcpConnection对象
void SignalingWorker::_newConn(int fd) {
    RTC_LOG(LS_INFO) << "signaling worker " << m_worker_id << ", receive fd: " << fd;


    if (fd < 0) {
        RTC_LOG(LS_WARNING) << "invalid fd: " << fd;
        return;
    }

    sock_setnonblock(fd);
    sock_setnodelay(fd);

    TcpConnection* conn = new TcpConnection(fd);
    sock_peer_to_str(fd, conn->m_ip, &(conn->m_port));
    // 创建并添加读事件
    conn->m_io_watcher = m_el->createIOEvent(conn_io_cb, this);
    m_el->startIOEvent(conn->m_io_watcher, fd, EventLoop::READ);
    // 创建并添加定时器事件
    conn->m_timer_watcher = m_el->createTimer(conn_time_cb, conn, true);
    m_el->startTimer(conn->m_timer_watcher, 100000); // 100ms
    conn->m_last_interaction = m_el->now();


    if ((size_t)fd >= m_connections.size()) {
        m_connections.resize(fd * 2, nullptr);
    }

    m_connections[fd] = conn;
}


// 处理TcpConnection对象的读事件
void SignalingWorker::_readQuery(int fd) {
    RTC_LOG(LS_INFO) << "signaling worker " << m_worker_id << " receive read event, fd: " << fd;

    if (fd < 0 || (size_t)fd >= m_connections.size()) {
        RTC_LOG(LS_WARNING) << "invalid fd: " << fd;
        return;
    }

    TcpConnection* conn = m_connections[fd];
    int nread = 0;
    int read_len = conn->m_bytes_expected;
    // 获取sds缓冲区实际的长度
    int qb_len = sdslen(conn->m_querybuf);
    // 确保 SDS 字符串 s至少有 addlen字节的​​空闲空间​​。若空间不足，则​​自动扩容​​
    conn->m_querybuf = sdsMakeRoomFor(conn->m_querybuf, read_len);
    nread = sock_read_data(fd, conn->m_querybuf + qb_len, read_len);

    conn->m_last_interaction = m_el->now();

    RTC_LOG(LS_INFO) << "sock read data, len: " << nread;
    
    if (nread == -1) {
        _closeConn(conn);
        return;
    } else if (nread > 0) {
        // 更新sds缓冲区内部长度信息
        sdsIncrLen(conn->m_querybuf, nread);
    }

    // 处理请求
    int ret = _processQueryBuffer(conn);
    if (ret != 0) {
        _closeConn(conn);
        return ;
    }

}

void SignalingWorker::_write_reply(int fd) {
    if (fd <= 0 || (size_t)fd >= m_connections.size()) {
        return;
    }

    TcpConnection* conn = m_connections[fd];
    if (!conn) {
        return;
    }

    while (!conn->m_reply_list.empty()) {
        rtc::Slice reply = conn->m_reply_list.front();
        int nwritten = sock_write_data(conn->m_fd, reply.data() + conn->m_cur_resp_pos, 
                                    reply.size() - conn->m_cur_resp_pos);

        if (nwritten == -1) {
            _closeConn(conn);
            return;
        } else if (nwritten == 0) {
            RTC_LOG(LS_WARNING) << "write zero bytes, fd: " << conn->m_fd << ", worker_id: " << m_worker_id;
        } else if (nwritten + conn->m_cur_resp_pos >= reply.size()) {
            // 写入完成
            conn->m_reply_list.pop_front();
            zfree((void*)reply.data());
            conn->m_cur_resp_pos = 0;
            RTC_LOG(LS_INFO) << "write finished, fd: " << conn->m_fd << ", worker_id: " << m_worker_id;
        } else {
            // 移动写入的字节
            conn->m_cur_resp_pos += nwritten;
        }
    }

    conn->m_last_interaction = m_el->now();
    // 写入队列为空，则停止对TcpConnection写事件的监听
    if (conn->m_reply_list.empty()) {
        m_el->stopIOEvent(conn->m_io_watcher, conn->m_fd, EventLoop::WRITE);
        RTC_LOG(LS_INFO) << "stop write event, fd: " << conn->m_fd << ", worker_id: " << m_worker_id;
    }
}


// 处理请求缓冲区中的消息
int SignalingWorker::_processQueryBuffer(TcpConnection* conn) {
    // 已完整读取一个完整的请求
    while (sdslen(conn->m_querybuf) >= conn->m_bytes_expected + conn->m_bytes_processed) {
        xhead_t* head = (xhead_t*)(conn->m_querybuf);
        // 校验魔数
        if (conn->m_current_state == TcpConnection::STATE_HEAD) {
            if (head->magic_num != XHEAD_MAGIC_NUM) {
                RTC_LOG(LS_WARNING) << "invalid data, fd: " << conn->m_fd;
                return -1;
            }

            conn->m_current_state = TcpConnection::STATE_BODY;
            conn->m_bytes_processed = XHEAD_SIZE;
            conn->m_bytes_expected = head->body_len;
        } else {
            // 正在读取请求数据
            rtc::Slice header(conn->m_querybuf, XHEAD_SIZE);
            rtc::Slice body(conn->m_querybuf + XHEAD_SIZE, conn->m_bytes_expected);

            int ret = _processRequest(conn, header, body);
            if (ret != 0) {
                return -1;
            }

            conn->m_bytes_processed = 65535;
        }
    }

    return 0;
}

int SignalingWorker::_processRequest(TcpConnection* conn, const rtc::Slice &header, const rtc::Slice &body) {
    RTC_LOG(LS_INFO) << "receive body: " << body.data();

    // 解析请求的json数据
    xhead_t* xh = (xhead_t*)(header.data());
    Json::CharReaderBuilder builder;
    std::unique_ptr<Json::CharReader> reader(builder.newCharReader());
    Json::Value root;
    JSONCPP_STRING errs;
    reader->parse(body.data(), body.data() + body.size(), &root, &errs);
    if (!errs.empty()) {
        RTC_LOG(LS_WARNING) << "parse json body error: " << errs << ", fd" << conn->m_fd
            << ", log_id: " << xh->log_id;
        return -1;
    }

    int cmdno, ret = 0;
    try {
        // 命令参数
        cmdno = root["cmdno"].asInt();
    } catch (const Json::Exception& e) {
        RTC_LOG(LS_WARNING) << "no cmdno field in body, log_id: " << xh->log_id;
        return -1;
    }
    
    // 处理命令参数
    switch (cmdno) {
        case CMDNO_PUSH:
            return _process_push(cmdno, conn, root, xh->log_id);
        case CMDNO_ANSWER:
            // 同步返回
            ret = _process_answer(cmdno, conn, root, xh->log_id);
            break;
        default:
            ret = -1;
            RTC_LOG(LS_WARNING) << "unknown cmdno: " << cmdno << ", log_id: " << xh->log_id;
            break;
    }

    // 返回处理结果
    char* buf = (char*)zmalloc(XHEAD_SIZE + MAX_RES_BUF);
    xhead_t* res_xh = (xhead_t*)buf;
    memcpy(res_xh, header.data(), header.size());

    Json::Value res_root;
    if (ret == 0) {
        res_root["err_no"] = 0;
        res_root["err_msg"] = "success";
    } else {
        res_root["err_no"] = -1;
        res_root["err_msg"] = "process error";
    }

    Json::StreamWriterBuilder writer_builder;
    writer_builder.settings_["indentation"] = "";
    std::string json_data = Json::writeString(writer_builder, res_root);
    RTC_LOG(LS_INFO) << "response body: " << json_data;

    res_xh->body_len = json_data.size();
    snprintf(buf + XHEAD_SIZE, MAX_RES_BUF, "%s", json_data.c_str());   

    rtc::Slice reply(buf, XHEAD_SIZE + res_xh->body_len);
    _add_reply(conn, reply);


    return 0;
}


// 解析推流请求，获取请求参数，向rtcserver的消息队列中投递消息
int SignalingWorker::_process_push(int cmdno, TcpConnection *c, const Json::Value &root, uint32_t log_id) {
    uint64_t uid;
    std::string stream_name;
    int audio, video;

    try {
        uid = root["uid"].asUInt64();
        stream_name = root["stream_name"].asString();
        audio = root["audio"].asInt();
        video = root["video"].asInt();
    } catch (const Json::Exception& e) {
        RTC_LOG(LS_WARNING) << "parse json body error: " << e.what() << "log_id: " << log_id;
        return -1;
    }

    RTC_LOG(LS_INFO) << "cmdno[" << cmdno << "] uid[" << uid 
        << "] stream_name[" << stream_name 
        << "] auido[" << audio 
        << "] video[" << video << "] signaling server push request";


    std::shared_ptr<RtcMsg> msg = std::make_shared<RtcMsg>();
    msg->m_cmdno = cmdno;
    msg->m_uid = uid;
    msg->m_stream_name = stream_name;
    msg->m_audio = audio;
    msg->m_video = video;        
    msg->m_log_id = log_id;
    msg->m_worker = this;
    msg->m_conn = c;
    msg->m_fd = c->m_fd;
    
    // 向rtcserver的消息队列中发送数据
    return g_rtc_server->sendRtcMsg(msg);
}

// 解析answer请求，获取请求参数，向rtcserver的消息队列中投递消息
int SignalingWorker::_process_answer(int cmdno, TcpConnection * /* c */, const Json::Value &root, uint32_t log_id)
{
    uint64_t uid;
    std::string stream_name;
    std::string answer;
    std::string stream_type;

    try {
        uid = root["uid"].asUInt64();
        stream_name = root["stream_name"].asString();
        answer = root["answer"].asString();
        stream_type = root["type"].asString();
    } catch (const Json::Exception& e) {
        RTC_LOG(LS_WARNING) << "parse json body error: " << e.what() << ", log_id: " << log_id;
        return -1;
    }

    RTC_LOG(LS_INFO) << "cmdno[" << cmdno << "] uid[" << uid 
        << "] stream_name[" << stream_name 
        << "] answer[" << answer 
        << "] stream_type[" << stream_type << "] signaling server send answer request";

    std::shared_ptr<RtcMsg> msg = std::make_shared<RtcMsg>();
    msg->m_cmdno = cmdno;
    msg->m_uid = uid;
    msg->m_stream_name = stream_name;
    msg->m_sdp = answer;
    msg->m_stream_type = stream_type;
    msg->m_log_id = log_id;

    return g_rtc_server->sendRtcMsg(msg);
}

// 处理RTCMSG
void SignalingWorker::_processRtcMsg() {
    std::shared_ptr<RtcMsg> msg = popMsg();
    if (!msg) {
        return;
    }

    switch (msg->m_cmdno) {
        case CMDNO_PUSH:
            _responseServerOffer(msg);
            break;
        default:
            RTC_LOG(LS_WARNING) << "unknown cmdno: " << msg->m_cmdno << ", log_id: " << msg->m_log_id;
            break;
    }
}

// 返回offer消息
void SignalingWorker::_responseServerOffer(std::shared_ptr<RtcMsg> msg) {
    TcpConnection* conn = (TcpConnection*)msg->m_conn;
    if (!conn) {
        return;
    }

    int fd = msg->m_fd;
    if (fd <= 0 || (size_t)fd >= m_connections.size()) {
        return ;
    }

    // 检查fd中的连接与之前透传的连接是否一致
    if (m_connections[fd] != conn) {
        return ;
    }

    // 构造响应头
    xhead_t* xh = (xhead_t*)conn->m_querybuf;
    rtc::Slice header(conn->m_querybuf, XHEAD_SIZE);
    char* buf = (char*)zmalloc(XHEAD_SIZE + MAX_RES_BUF);
    if (!buf) {
        RTC_LOG(LS_WARNING) << "zmalloc error, log_id: " << xh->log_id;
        return;
    }

    memcpy(buf, header.data(), XHEAD_SIZE);
    xhead_t* res_xh = (xhead_t*)buf;

    try {
        Json::Value res_root;
        res_root["err_no"] = msg->m_err_no;
        if (msg->m_err_no != 0) {
            res_root["err_msg"] = "process error    ";
            res_root["offer"] = "";
        } else {
            res_root["err_msg"] = "success";
            res_root["offer"] = msg->m_sdp;
        }

        Json::StreamWriterBuilder builder;
        // 设置json无缩进
        builder.settings_["indentation"] = "";
        std::string json_data = Json::writeString(builder, res_root);
        RTC_LOG(LS_INFO) << "response body: " << json_data;

        res_xh->body_len = json_data.size();
        snprintf(buf + XHEAD_SIZE, MAX_RES_BUF, "%s", json_data.c_str());
        
        rtc::Slice reply(buf, XHEAD_SIZE + res_xh->body_len);
        _add_reply(conn, reply);
    } catch (const Json::Exception& e) {
        RTC_LOG(LS_WARNING) << "create json response error: " << e.what() << ", log_id: " << xh->log_id;
    }    
}

// 将offer响应包写入tcp连接
void SignalingWorker::_add_reply(TcpConnection *conn, const rtc::Slice &reply) {
    conn->m_reply_list.push_back(reply);
    // 添加写事件
    m_el->startIOEvent(conn->m_io_watcher, conn->m_fd, EventLoop::WRITE);
}

// 处理tcp连接超时MAX_RES_BUF
void SignalingWorker::_processTimeout(TcpConnection* conn) {
    if (m_el->now() - conn->m_last_interaction >= (unsigned long)m_options.m_connection_timeout) {
        RTC_LOG(LS_INFO) << "connection timeout, fd: " << conn->m_fd; 
        _closeConn(conn);
    }
}

// 关闭tcp连接
void SignalingWorker::_closeConn(TcpConnection* conn) {
    RTC_LOG(LS_INFO) << "close connection, fd: " << conn->m_fd;
    close(conn->m_fd);
    _removeConn(conn);
}


// 移除tcp连接对象
void SignalingWorker::_removeConn(TcpConnection* conn) {
    m_el->deleteTimer(conn->m_timer_watcher);
    m_el->deleteIOEvent(conn->m_io_watcher);
    m_connections[conn->m_fd] = nullptr;
    delete conn;
}

void SignalingWorker::_stop()
{
    if (m_thread == nullptr) {
        RTC_LOG(LS_WARNING) << "signaling worker not running, worker_id: " << m_worker_id;
        return;
    }

    m_el->deleteIOEvent(m_pipe_watcher);
    m_el->stop();

    close(m_notify_recv_fd);
    close(m_notify_send_fd);
}

} // namespace xrtc
